package edu.hhu.wifispark.task.sparktask;

import java.io.Serializable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;

import edu.hhu.wifispark.entry.BasePhone;
import edu.hhu.wifispark.entry.Record;
import edu.hhu.wifispark.util.SparkProUtil;
import scala.Tuple2;

public class RecordTask extends Thread implements Serializable{
	private static final long serialVersionUID = 1L;
	public static String ZERO = "0";
	public static String ONE = "1";
	
	public List<BasePhone> bps;
	String taskName = null;
	public RecordTask(){
		this.taskName = "record";
	}
	public RecordTask(long time){
		this.taskName = "record" + time;
	}
	
	@SuppressWarnings("resource")
	public void run() {
		SparkConf sparkConf = new SparkConf().setAppName(taskName);
		sparkConf.setMaster(SparkProUtil.MASTER);
		JavaSparkContext jsc = new JavaSparkContext(sparkConf);
		JavaRDD<BasePhone> phoneRDD = jsc.parallelize(bps);
		
		JavaPairRDD<String,String> phoneStrRdd = phoneRDD.mapToPair(new PairFunction<BasePhone, String, String>(){
			private static final long serialVersionUID = 1L;

			public Tuple2<String, String> call(BasePhone basePhone){
				String mac = basePhone.getStr("mac");
				String value = basePhone.getLong("atime")+","+basePhone.getInt("inner");
				return new Tuple2<String, String>(mac, value);
			}
		});
		
		JavaPairRDD<String, Iterable<String>> phoneGroup =  phoneStrRdd.groupByKey();
		JavaPairRDD<String,String> phoneSeq = phoneGroup.mapToPair(new PairFunction<Tuple2<String,Iterable<String>>, String, String>() {
			private static final long serialVersionUID = 1L;

			public Tuple2<String, String> call(Tuple2<String, Iterable<String>> t) {
				Iterator<String> its = t._2.iterator();
				StringBuffer sb = new StringBuffer();
				String timeStr = null;
				String timeEnd = null;
				String isInner = "0";
				String[] values;
				while(its.hasNext()){
					values = its.next().split(",");
					timeEnd = values[0];
					if(ZERO.equals(isInner) && ONE.equals(values[1])){
						isInner = "1";
						timeStr = timeEnd;
					}else if (ONE.equals(isInner) && ZERO.equals(values[1])) {
						isInner = "0";
						sb.append(timeStr+"-"+timeEnd+",");
					}
				}
				if(ONE.equals(isInner) && !timeEnd.equals(timeStr)){
					sb.append(timeStr+"-"+timeEnd);
				}else{
					if(sb.length() > 0){
						sb.deleteCharAt(sb.length() - 1);
					}
				}
				return new Tuple2<String, String>(t._1, sb.toString());
			}
		});
		Map<String,String> maps = phoneSeq.collectAsMap();
		jsc.stop();
		for(Map.Entry<String, String> entry : maps.entrySet()){
			String mac = entry.getKey();
			String values = entry.getValue();
			if(values==null || "".equals(values)){
				continue;
			}
			String[] items = values.split(",");
			for(String item : items){
				String[] timese = item.split("-");
				long entertime = Long.parseLong(timese[0]);
				long outtime = Long.parseLong(timese[1]);
				long staytime = outtime - entertime;
				Record record = new Record();
				record.set("mac",mac);
				record.set("entertime", entertime);
				record.set("outtime",outtime);
				record.set("staytime", staytime);
				record.save();
			}
		}
//		System.out.println(maps.keySet());
	}
	
	public void run0(){
		String mac = "";
		int inner = 3;
		int flag = 0;
		long entertime = 0;
		long outtime = 0, staytime;
		
		for(BasePhone str:bps){
			Record record = new Record();
			if(!mac.equals(str.getStr("mac"))){
				mac = str.getStr("mac");
				inner = str.getInt("inner");
				flag = 0;
								
			}
			else 
			{
				if(str.getInt("inner") == 1){
					if(inner == 0){
						//有入店行为
						entertime = str.getLong("atime");
						flag = 1;
					}
					inner = 1;
				}
				else 
				{
					if(inner == 1 && flag ==1){
						//有出店行为且之前有入店行为
						outtime = str.getLong("atime");
						staytime = outtime - entertime;
						record.set("mac",mac);
						record.set("entertime", entertime);
						record.set("outtime",outtime);
						record.set("staytime", staytime);
						record.save();
						flag = 2;
					}
					inner = 0;
				}
			}
		}
	}
	
	public void setBps(List<BasePhone> ibps){
		this.bps = ibps;
	}
}
